package com.techsun.handler;

import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.http.HTTPSourceHandler;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;

import javax.servlet.http.HttpServletRequest;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.*;

/**
 * 将不同租户数据发往不同 top , 规则: retailevent 和 customer 加上 租户id的后缀
 */
public class TenantByIdHTTPHandler implements HTTPSourceHandler {
    Logger logger = Logger.getLogger(TenantByIdHTTPHandler.class);
    public static final String charset = "UTF-8";
    private String bootstrapServers;
    private int pollIntervalSecond;
    private Map existTopic;

    @Override
    public void configure(Context context) {
        bootstrapServers = context.getString("bootstrapServers");
        pollIntervalSecond = Integer.parseInt(context.getString("pollIntervalSecond", "10"));
        getTopicList();
    }

    private void getTopicList() {
        new Thread(() -> {
            while (true) {
                KafkaConsumer kafkaConsumer = null;
                try {
                    Properties props = new Properties();
                    props.put("bootstrap.servers", bootstrapServers);
                    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
                    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
                    kafkaConsumer = new KafkaConsumer<String, String>(props);
                    existTopic = kafkaConsumer.listTopics();
                    Thread.sleep(pollIntervalSecond * 1000L);
                } catch (Exception e) {
                    logger.error("get topic map has error: ", e);
                } finally {
                    if (kafkaConsumer != null) {
                        kafkaConsumer.close();
                    }
                }
            }
        }).start();
    }


    @Override
    public List<Event> getEvents(HttpServletRequest request) {
        StringBuilder sb = new StringBuilder();
        List<Event> events = new LinkedList<>();
        try {
            if (request.getMethod().equalsIgnoreCase("POST")) {
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(request.getInputStream(), charset));
                String inputStr;
                while ((inputStr = bufferedReader.readLine()) != null) {
                    sb.append(inputStr);
                }
                logger.error(String.format("==== POST request ==== \n host:[%s] body:[%s]", request.getRemoteHost(), sb));

                JSONObject jsonObject = JSONObject.parseObject(sb.toString());
                // 获取租户id
                String project = jsonObject.getString("project").trim();
                String customerTopic = "customer_" + project;
                String retailEventTopic = "retailevent_" + project;
                if (!jsonObject.containsKey("posted_time")) {
                    jsonObject.put("posted_time", System.currentTimeMillis());
                }
                Map<String, String> map = new HashMap<>();
                // 判断注册事件的topic是否存在
                if ("track_signup".equalsIgnoreCase(jsonObject.getString("event_type"))) {
                    if (existTopic.containsKey(customerTopic)) {
                        map.put("topic", customerTopic);
                        events.add(EventBuilder.withBody(jsonObject.toString().getBytes(charset), map));
                    } else {
                        logger.error(String.format("==== this topic: [%s] is not exist ====", customerTopic));
                    }
                }
                // 判断行为事件的topic是否存在
                if (existTopic.containsKey(retailEventTopic)) {
                    map.put("topic", retailEventTopic);
                    map.put("key", jsonObject.getString("customer_id"));
                    events.add(EventBuilder.withBody(jsonObject.toString().getBytes(charset), map));
                } else {
                    logger.error(String.format("==== this topic: [%s] is not exist ====", retailEventTopic));
                }
                return events;
            }
        } catch (Exception e) {
            logger.error(String.format("this request: [%s] parse data exception: [%s]", sb, e));
        }
        return events;
    }

}
